
#include "RtCnRudpBase.h"
#include "RtCnRudp.h"
#include "RtCnRudpFlowControl.h"
#include "RtObserver.h"
#include <malloc.h>

static const int DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE = 128;
static const float PACKETLOSS_TOLERANCE = .04f; // What percentile packetloss we are willing to accept as background noise.
#if 0
static const double MINIMUM_SEND_BPS = 14400.0; // Won't go below this send rate
static const double STARTING_SEND_BPS = 28800.0; // What send rate to start at.
#else
static const double MINIMUM_SEND_BPS = 57344; // Won't go below this send rate
static const double STARTING_SEND_BPS = 10485760*10; // What send rate to start at.
//static const double STARTING_SEND_BPS = 114688; // What send rate to start at.
#endif
static const RtRudpTimeUS TIME_TO_NEW_SAMPLE = 500000; // How many ns to wait before starting a new sample.  This way buffers have time to overflow or relax at the new send rate, if they are indeed going to overflow.
static const RtRudpTimeUS MAX_TIME_TO_SAMPLE = 250000; // How many ns to sample the connection before deciding on a course of action(increase or decrease throughput). You must be at full send rate the whole time

static const RtRudpTime MAX_PING_VALUE = 1500; 
static const int PING_MULTIPLIER_TO_RESEND = 3; // So internet ping variation doesn't cause needless resends

/// Maximum (stack) size to use with _alloca before using new and delete instead.
#define MAX_ALLOCA_STACK_ALLOCATION 1048576

#ifndef DEFAULT_MTU_SIZE

/// The MTU size to use.
/// \li \em 17914 16 Mbit/Sec Token Ring
/// \li \em 4464 4 Mbits/Sec Token Ring
/// \li \em 4352 FDDI
/// \li \em 1500. The largest Ethernet packet size \b recommended. This is the typical setting for non-PPPoE, non-VPN connections. The default value for NETGEAR routers, adapters and switches.
/// \li \em 1492. The size PPPoE prefers.
/// \li \em 1480. WinXP PPPoE. (test on CNC PPPoE)
/// \li \em 1472. Maximum size to use for pinging. (Bigger packets are fragmented.)
/// \li \em 1468. The size DHCP prefers.
/// \li \em 1460. Usable by AOL if you don't have large email attachments, etc.
/// \li \em 1430. The size VPN and PPTP prefer.
/// \li \em 1400. Maximum size for AOL DSL.
/// \li \em 1396. WinXP VPN.
/// \li \em 576. Typical value to connect to dial-up ISPs.
#define DEFAULT_MTU_SIZE 1200

/// The largest value for an UDP datagram
#define MAXIMUM_MTU_SIZE 1500

#endif

// #define RUDP_SEND_BUF_SIZE_DEFAULT 1048576 // in bytes  
#define RUDP_SEND_BUF_SIZE_DEFAULT 8192 // in bytes  

/// Sizeof an UDP header in byte
#define UDP_HEADER_SIZE 28

#define RUDP_SEND_SYNC_INTERVAL 1 // in seconds
#define RUDP_VERSION 0x01
#define RUDP_MAGIC_NUMBER 305419896 
#define RUDP_CONNREQ_LEN (sizeof(BYTE)+sizeof(DWORD)*2+sizeof(WORD))

#define RUDP_CONNECTION_TIMEOUT 180 // in seconds

#define RUDP_KEEPALIVE_INTERVAL 15 // in seconds

#define RUDP_NOT_MORE_THAN(a, b, type) ( (type) ( (type)(b)-(type)(a) )<=((type)-1)/2 )
#define RUDP_IS_RELIABLE(reliability) ( reliability==RUDP_RELIABLE || reliability==RUDP_RELIABLE_SEQUENCED || reliability==RUDP_RELIABLE_ORDERED )

#ifndef RT_LINUX
	/// Send immediately only on client platform, 
	/// while for servers under LINUX, the overall performance is the first consideration. 
	#define RUDP_SEND_IMMEDIATELY 
#endif

/// Threshold of the dropped packets above which we would like to increase the resending times, 
/// about 5% of all packets, promising that 1 audio packet is sent out every 30 milliseconds.
static const WORD s_dropThresholdForResend = RUDP_KEEPALIVE_INTERVAL*1000/30*5/100;
static const WORD s_dropThresholdForTick = s_dropThresholdForResend*3;
static const WORD s_maxDropTick = 2;

#ifdef RUDP_ENABLE_CHECK_PACKETNO
static CRtString s_recvPacketNoList;
static CRtString s_sendPacketNoList;
#endif // RUDP_ENABLE_CHECK_PACKETNO

#define RUDP_MAX_RESEND_INCREMENT 2

int SplitPacketIndexComp( RtRudpSplitPacketID  &key, RtRudpPacket*  &data )
{
	if (key < data->splitPacketIndex)
		return -1;
	if (key == data->splitPacketIndex)
		return 0;
	return 1;
}

int SplitPacketChannelComp( RtRudpSplitPacketNo  &key, SplitPacketChannel*  &data )
{
	if (key < data->splitPacketList[0]->splitPacketId)
		return -1;
	if (key == data->splitPacketList[0]->splitPacketId)
		return 0;
	return 1;
}

//////////////////////////////////////////////////////////////////////////
// class CRtRudpConn
//////////////////////////////////////////////////////////////////////////

CRtRudpConn::CRtRudpConn()
: m_mtuSize(DEFAULT_MTU_SIZE)
, m_maxSendBufferSize(RUDP_SEND_BUF_SIZE_DEFAULT)
, m_totalSizeOfPassedInData(0)
, m_totalSizeOfSentOutData(0)
, m_needKeepAlive(FALSE)
, m_needOnSend(FALSE)
#ifdef RUDP_FLOWCONTROL_OPTIMIZE
, m_onFlowControl(FALSE)
#endif // RUDP_FLOWCONTROL_OPTIMIZE
, m_isAckSent(FALSE)
, m_isDataSent(FALSE)
, m_recvDropPackets(0)
, m_sendDropPackets(0)
, m_recvDropTick(0)
, m_sendDropTick(0)
, m_observer(NULL)
, m_resendIncrement(0)
//, m_lastWarningResendQueueTime(0)
, updateBitStream( MAXIMUM_MTU_SIZE-UDP_HEADER_SIZE )
, timeoutTime(RUDP_CONNECTION_TIMEOUT*1000000)
{
	m_status = RUDP_STATE_INIT;

	m_pThreadNetwork = CRtThreadManager::Instance()->GetThread(CRtThreadManager::TT_NETWORK);
	m_pFlowControl = CRtRudpFlowControl::Instance();

	InitializeVariables();
	AddReference();
}

CRtRudpConn::~CRtRudpConn()
{
	Release();
}

RtResult CRtRudpConn::SetOption(DWORD aCommand, LPVOID aArg)
{
	switch (aCommand) 
	{
	case CS_OPT_MAX_SENDBUF_LEN:
		m_maxSendBufferSize = *static_cast<DWORD*>(aArg);
		RUDP_INFO_TRACE_THIS("CRtRudpConn::SetOption, maxSendBufferSize = " << m_maxSendBufferSize);
		break;
	case RT_OPT_TRANSPORT_MTU:
		m_mtuSize = *static_cast<INT*>(aArg);
		RUDP_INFO_TRACE_THIS("CRtRudpConn::SetOption, mtuSize = " << m_mtuSize);
		break;
	case RT_OPT_OBSERVER:
		m_observer = static_cast<IRtObserver*>(aArg);
		RUDP_INFO_TRACE_THIS("CRtRudpConn::SetOption, observer = " << m_observer);
		break;
	default:
		if(m_lowerTransport)
			return m_lowerTransport->SetOption(aCommand, aArg);
		return RT_ERROR_NOT_AVAILABLE;
	}
	return RT_OK;
}

RtResult CRtRudpConn::GetOption(DWORD aCommand, LPVOID aArg)
{
	switch (aCommand) 
	{
	case RT_OPT_TRANSPORT_TRAN_TYPE:
		*(static_cast<CRtConnectionManager::CType*>(aArg)) = CRtConnectionManager::CTYPE_RUDP;
		break;
	case CS_OPT_MAX_SENDBUF_LEN:
		*(static_cast<DWORD*>(aArg)) = m_maxSendBufferSize;
		break;
	case RT_OPT_TRANSPORT_MTU:
		*(static_cast<INT*>(aArg)) = m_mtuSize;
		break;
#ifdef RUDP_ENABLE_STATISTICS
	case RT_OPT_RUDP_STATISTICS:
		{
			unsigned i=0;

			m_statistics.m_messageNumber = messageNumber;
			m_statistics.m_receivedPacketsBaseIndex = receivedPacketsBaseIndex;
			m_statistics.m_curBandwidth = currentBandwidth;
			m_statistics.m_packetsOnResendQueue = resendList.Size();

			m_statistics.m_packetsWaitingForReorder = 0;
			for ( i = 0; i < orderingList.Size(); i++ )
			{
				if (orderingList[i])
				{
					m_statistics.m_packetsWaitingForReorder += orderingList[i]->Size();
				}
			}

			m_statistics.m_splitsWaitingForReassembly = 0;
			for (i=0; i < splitPacketChannelList.Size(); i++)
				m_statistics.m_splitsWaitingForReassembly += splitPacketChannelList[i]->splitPacketList.Size();

			*(static_cast<CRtRudpStatistics*>(aArg)) = m_statistics;
		}
		break;
#endif
	default:
		if(m_lowerTransport)
			return m_lowerTransport->GetOption(aCommand, aArg);
		return RT_ERROR_NOT_AVAILABLE;
	}
	return RT_OK;
}

RtResult CRtRudpConn::Disconnect(RtResult aReason)
{
	if (m_status==RUDP_STATE_CLOSED)
		return RT_OK;

	RUDP_INFO_TRACE_THIS("CRtRudpConn::Disconnect, aReason=" << aReason	<< " state=" << m_status);

	m_sink = NULL;

	if (m_status!=RUDP_STATE_OPEN)
	{
		SetStatus(RUDP_STATE_CLOSED);
	 	Release();
	 	ReleaseReference();
		return RT_OK;
	}

	RtRudpTimeUS time = GetRtRudpTimeUS();

	// Make sure all data be sent out before disconnected.
	WrapUpPdusToPackets(/*time*/);

	CRtMessageBlock mb(0);
	SendPdu((RtRudpPduType)RUDP_PDU_SHUTDOWN, mb, RUDP_RELIABLE_ORDERED, DP_LOW);
	UpdateSending(time);

	SetStatus(RUDP_STATE_CLOSED);
#if 1
	/// To release socket resources, we have to call Release() immediately.
	Release();
	ReleaseReference();
	return RT_OK;
#else
	CRtTimeValue tv( (time_t)0, (LONG) ( (ping<50) ? ping*2 : 100 ) * 1000 ); // Wait at most 100 milliseconds. 
	return m_timer.Schedule(this, tv, 1);
#endif
}

void CRtRudpConn::OnReceive_i(
		CRtMessageBlock &aData,
		IRtTransport *aTrptId,
		CRtTransportParameter *aPara)
{
	DWORD dataLen = aData.GetChainedLength();

	if (dataLen!=aData.GetTopLevelLength())
	{
		RUDP_ERROR_TRACE_THIS("CRtRudpConn::OnReceive_i, unexpected! chainlen="<<dataLen<<" topLen="<<aData.GetTopLevelLength());
		return;
	}

#ifdef RUDP_ENABLE_STATISTICS
	m_statistics.m_totalBytesReceived += dataLen;
	++m_statistics.m_packagesReceived;
#endif

#if 1 // For performance.
	RtRudpTimeUS time = m_lastPacketReceiveTime = lastUpdateTime;
#else
	RtRudpTimeUS time = m_lastPacketReceiveTime = GetRtRudpTimeUS();
#endif

	RtRudpPacketNo holeCount;
	unsigned i;
	bool hasAcks=false;

	RakNet::BitStream socketData( (unsigned char*) aData.GetTopLevelReadPtr(), dataLen, false ); 

	DataStructures::RangeList<RtRudpPacketNo> incomingAcks;
	socketData.Read(hasAcks);
	if (hasAcks)
	{
		RtRudpPacketNo packetNo;
		if (incomingAcks.Deserialize(&socketData)==false)
			return;

		RtRudpTime rtt = 0;
		RtRudpTime totalRttVal = ping;
		unsigned short totalRttNum = 1;

#ifdef RUDP_ENABLE_CHECK_PACKETNO
		CRtString strAckPacketNo;
		char szPacketNo[64] = "";
#endif // RUDP_ENABLE_CHECK_PACKETNO

		for (i=0; i<incomingAcks.ranges.Size(); i++)
		{
			if (incomingAcks.ranges[i].minIndex>incomingAcks.ranges[i].maxIndex)
			{
				RT_ASSERTE(incomingAcks.ranges[i].minIndex<=incomingAcks.ranges[i].maxIndex);
				return;
			}

#ifdef RUDP_ENABLE_CHECK_PACKETNO
			snprintf(szPacketNo, 63, "[%u,%u]", incomingAcks.ranges[i].minIndex, incomingAcks.ranges[i].maxIndex);
			strAckPacketNo.append(szPacketNo);
#endif // RUDP_ENABLE_CHECK_PACKETNO

			for (packetNo=incomingAcks.ranges[i].minIndex; packetNo >= incomingAcks.ranges[i].minIndex && packetNo <= incomingAcks.ranges[i].maxIndex; packetNo++)
			{
#ifdef RUDP_ENABLE_BW_DETECT
				unsigned ackedHistogramCounter=
#endif
					RemovePacketFromResendList( packetNo, time, rtt );

				if (rtt>0 && rtt<ping*PING_MULTIPLIER_TO_RESEND)
				{
					totalRttVal += rtt;
					++totalRttNum;
				}

#ifdef RUDP_ENABLE_BW_DETECT
				if (time >= histogramStartTime && ackedHistogramCounter!=(unsigned)-1 && ackedHistogramCounter==histogramReceiveMarker)
					++histogramAckCount;
#endif // RUDP_ENABLE_BW_DETECT

				if ( resendList.IsEmpty() )
				{
					lastAckTime = 0; // Not resending anything so clear this var so we don't drop the connection on not getting any more acks
				}
				else
				{
					lastAckTime = time; // Just got an ack.  Record when we got it so we know the connection is alive
				}
			}
		}

#ifdef RUDP_ENABLE_CHECK_PACKETNO
		if (!strAckPacketNo.empty())
		{
			RUDP_INFO_TRACE_THIS("ACK PacketNo="<<strAckPacketNo);
		}
#endif // RUDP_ENABLE_CHECK_PACKETNO

		if (m_status==RUDP_STATE_CLOSED && resendList.IsEmpty())
		{
			// The peer endpoint has acknowledged the shutdown pdu and all other data.
			// Thus, the connection can be gracefully closed now.
			Release();
			ReleaseReference();
			return;
		}

		if (totalRttNum>1)
		{
			SetPing(totalRttVal/totalRttNum);
		}

		CheckOnSend();
	}


	// Parse the bitstream to create an internal packet
	RtRudpPacket* internalPacket = CreateInternalPacketFromBitStream( &socketData, time );

	while ( internalPacket )
	{
		{
			if ( RUDP_IS_RELIABLE(internalPacket->reliability) )
			{
#ifdef RUDP_ENABLE_CHECK_PACKETNO
				static char s_recvPacketNo[64] = "";
				snprintf(s_recvPacketNo, 63, "%u ", internalPacket->packetNo);
				s_recvPacketNoList.append(s_recvPacketNo);
				if (s_recvPacketNoList.length()>1900)
				{
					RUDP_INFO_TRACE_THIS("RECV PacketNo List: "<<s_recvPacketNoList);
					s_recvPacketNoList = "";
				}
#endif // RUDP_ENABLE_CHECK_PACKETNO

				SendAcknowledgementPacket( internalPacket->packetNo, time );

				// If the following conditional is true then this either a duplicate packet
				// or an older out of order packet
				// The subtraction unsigned overflow is intentional
				holeCount = (RtRudpPacketNo)(internalPacket->packetNo-receivedPacketsBaseIndex);
				const int typeRange = (RtRudpPacketNo)-1;

				if (holeCount==0)
				{
					// Got what we were expecting
					if (hasReceivedPacketQueue.Size())
						hasReceivedPacketQueue.Pop();
					++receivedPacketsBaseIndex;
				}
				else if (holeCount > typeRange/2)
				{
					// Duplicate packet
//					internalPacket->data->DestroyChained();
					m_pFlowControl->m_internalPacketPool->Delete( internalPacket );

#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_duplicatePacketsReceived;
#endif

					goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
				}
				else if (holeCount<hasReceivedPacketQueue.Size())
				{
					// Got a higher count out of order packet that was missing in the sequence or we already got
					if (hasReceivedPacketQueue[holeCount]!=0) // non-zero means this is a hole
					{
						// Fill in the hole
						hasReceivedPacketQueue[holeCount]=0; // We got the packet at holeCount
					}
					else
					{
						// Duplicate packet
//						internalPacket->data->DestroyChained();
						m_pFlowControl->m_internalPacketPool->Delete( internalPacket );

#ifdef RUDP_ENABLE_STATISTICS
						++m_statistics.m_duplicatePacketsReceived;
#endif

						goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
					}
				}
				else // holeCount>=receivedPackets.Size()
				{
					// Got a higher count out of order packet whose packetNo is higher than we have ever got

					// Add 0 times to the queue until (packetNo - baseIndex) < queue size.
					while ((RtRudpPacketNo)(holeCount) > hasReceivedPacketQueue.Size())
						hasReceivedPacketQueue.Push(time+10000000); // Didn't get this packet - set the time to give up waiting
					hasReceivedPacketQueue.Push(0); // Got the packet

/*
					// If this assert hits then RtRudpPacketNo has overflowed
					RT_ASSERTE(hasReceivedPacketQueue.Size() < (unsigned int)((RtRudpPacketNo)(-1)));
*/
				}

#if 0
				if (internalPacket->pduType==RUDP_PDU_DROPPACKET)
				{
					HandleDropPacket(internalPacket, time);

#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_dropPacketsReceived;
#endif

					goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
				}
#endif

				ReceivedPacketQueuePopExpiredTimeAndCompress(time);
			}

#ifdef RUDP_ENABLE_STATISTICS
			++m_statistics.m_packetsReceived;
#endif

			// Keep on top of deleting old unreliable split packets so they don't clog the list.
			if ( internalPacket->splitPacketCount > 0 )
				DeleteOldUnreliableSplitPackets( time );

			if ( internalPacket->reliability == RUDP_RELIABLE_SEQUENCED || internalPacket->reliability == RUDP_UNRELIABLE_SEQUENCED )
			{
				if ( internalPacket->orderingChannel >= NUMBER_OF_ORDERED_STREAMS )
				{
					// Invalid packet
					RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnReceive, Got invalid packet");
//					internalPacket->data->DestroyChained();
					m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
					goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
				}

				if ( IsOlderOrderedPacket( internalPacket->orderingIndex, waitingForSequencedPacketReadIndex[ internalPacket->orderingChannel ] ) )
				{
					// Older sequenced packet. Discard it
//					internalPacket->data->DestroyChained();
					m_pFlowControl->m_internalPacketPool->Delete( internalPacket );

#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_sequencedPacketsOutOfOrder;
#endif
				}
				else
				{
#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_sequencedPacketsInOrder;
#endif

					// Is this a split packet?
					if ( internalPacket->splitPacketCount > 0 )
					{
						// Check for a rebuilt packet
						InsertIntoSplitPacketList( internalPacket, time );

						// Sequenced
						internalPacket = BuildPacketFromSplitPacketList( internalPacket->splitPacketId, time );
					}

					if ( internalPacket )
					{
						// Update our index to the newest packet
						waitingForSequencedPacketReadIndex[ internalPacket->orderingChannel ] = internalPacket->orderingIndex + 1;

						HandleReceivedPacket(internalPacket);
					}
				}

				goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
			}

			// Is this an unsequenced split packet?
			if ( internalPacket->splitPacketCount > 0 )
			{
				// An unsequenced split packet.  May be ordered though.

				// Check for a rebuilt packet

				InsertIntoSplitPacketList( internalPacket, time );

				internalPacket = BuildPacketFromSplitPacketList( internalPacket->splitPacketId, time );

				if ( internalPacket == 0 )
				{
					// Don't have all the parts yet
					goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
				}

				// else continue down to handle RUDP_RELIABLE_ORDERED
			}

			if ( internalPacket->reliability == RUDP_RELIABLE_ORDERED )
			{
				if ( internalPacket->orderingChannel >= NUMBER_OF_ORDERED_STREAMS )
				{
					// Invalid packet
					RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnReceive, Got invalid ordering channel "<<internalPacket->orderingChannel<<" from packet "<<internalPacket->packetNo);
//					internalPacket->data->DestroyChained();
					m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
					goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
				}

				if ( waitingForOrderedPacketReadIndex[ internalPacket->orderingChannel ] == internalPacket->orderingIndex )
				{
#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_orderedPacketsInOrder;
#endif

					// Get the list to hold ordered packets for this stream
					RudpRecvOrderingListType *orderingListAtOrderingStream;
					unsigned char orderingChannelCopy = internalPacket->orderingChannel;

					HandleReceivedPacket(internalPacket);

					// Wait for the next ordered packet in sequence
					waitingForOrderedPacketReadIndex[ orderingChannelCopy ] ++; // This wraps

					orderingListAtOrderingStream = GetOrderingListAtOrderingStream( orderingChannelCopy );

					if ( orderingListAtOrderingStream != 0)
					{
						while ( orderingListAtOrderingStream->Size() > 0 )
						{
							// Cycle through the list until nothing is found
							orderingListAtOrderingStream->Beginning();
							bool indexFound=false;
							int size=orderingListAtOrderingStream->Size();
							int count=0;

							while (count++ < size)
							{
								if ( orderingListAtOrderingStream->Peek()->orderingIndex == waitingForOrderedPacketReadIndex[ orderingChannelCopy ] )
								{
									HandleReceivedPacket(orderingListAtOrderingStream->Pop());
									waitingForOrderedPacketReadIndex[ orderingChannelCopy ]++; // This wraps at 65535
									indexFound=true;
								}
								else
									(*orderingListAtOrderingStream)++;
							}

							if (indexFound==false)
								break;
						}
					}
				}
				else
				{
#ifdef RUDP_ENABLE_STATISTICS
					++m_statistics.m_orderedPacketsOutOfOrder;
#endif
//					RT_ASSERTE(waitingForOrderedPacketReadIndex[ internalPacket->orderingChannel ] < internalPacket->orderingIndex);
					if (IsOlderOrderedPacket(internalPacket->orderingIndex, waitingForOrderedPacketReadIndex[ internalPacket->orderingChannel ]))
					{
						RUDP_ERROR_TRACE_THIS("CRtRudpConn::OnReceive_i, waitingOrderId "<<waitingForOrderedPacketReadIndex[ internalPacket->orderingChannel ]
							<<" while getting "<<internalPacket->orderingIndex
							<<" waitingPacketId "<<receivedPacketsBaseIndex
							<<" and getting "<<internalPacket->packetNo);
					}

					// This is a newer ordered packet than we are waiting for. Store it for future use
					AddToOrderingList( internalPacket );
				}

				goto CONTINUE_SOCKET_DATA_PARSE_LOOP;
			}

			HandleReceivedPacket(internalPacket);
		}

		// Used for a goto to jump to the next packet immediately

CONTINUE_SOCKET_DATA_PARSE_LOOP:
		// Parse the bitstream to create an internal packet
		internalPacket = CreateInternalPacketFromBitStream( &socketData, time );
	}

#ifdef RUDP_FLOWCONTROL_OPTIMIZE

	if (InsertIntoFlowControl())
		return;

	RemoveFromFlowControl();

#endif // RUDP_FLOWCONTROL_OPTIMIZE
}

void CRtRudpConn::OnSend(
		IRtTransport *aTrptId,
		CRtTransportParameter *aPara)
{
	RUDP_INFO_TRACE_THIS("CRtRudpConn::OnSend");
	RT_ASSERTE(FALSE);
}

void CRtRudpConn::OnTimer(CRtTimerWrapper* aId)
{
	RT_ASSERTE(aId==&m_timer);

	if (m_status==RUDP_STATE_OPEN)
	{
#ifdef RUDP_ENABLE_CHECK_PACKETNO
		if (!s_sendPacketNoList.empty())
		{
			RUDP_INFO_TRACE_THIS("SEND PacketNo List: "<<s_sendPacketNoList);
			s_sendPacketNoList = "";
		}
		if (!s_recvPacketNoList.empty())
		{
			RUDP_INFO_TRACE_THIS("RECV PacketNo List: "<<s_recvPacketNoList);
			s_recvPacketNoList = "";
		}
#endif // RUDP_ENABLE_CHECK_PACKETNO

		RtRudpTimeUS time = lastUpdateTime; // GetRtRudpTimeUS();
		if (time - m_lastPacketReceiveTime > timeoutTime && time > m_lastPacketReceiveTime)
		{
			// dead connection
			// We've waited a very long time for a peer packet to arrive. 
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, connection lost! Long time no packet!");
			OnDisconnect(RT_ERROR_NETWORK_CONNECTION_LOST, m_lowerTransport.ParaIn());
			return;
		}

		if (RUDP_SEND_BUFFER_SIZE>3072)
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, BuffSize="<<RUDP_SEND_BUFFER_SIZE);

		for (unsigned channel = 0; channel<orderingList.Size(); ++channel)
		{
			if (orderingList[channel]!=NULL && orderingList[channel]->Size()>33)
			{
				RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, too many out-of-ordered packets! channel="<<channel<<" size="<<orderingList[channel]->Size());
			}
		}

		// Send keep-alive pdu.
		if (m_needKeepAlive)
		{
			if (!m_isDataSent)
			{
				CRtMessageBlock mb(0);
				SendPdu((RtRudpPduType)RUDP_PDU_KEEPALIVE, mb, RUDP_RELIABLE, DP_LOW);
#ifdef RUDP_FLOWCONTROL_OPTIMIZE
				InsertIntoFlowControl();
#endif // RUDP_FLOWCONTROL_OPTIMIZE
			}
			else
				m_isDataSent = FALSE;
		}

		BOOL callObserver = FALSE;
		BOOL clearRecvDropTick = TRUE;
		BOOL clearSendDropTick = TRUE;

		if (m_recvDropPackets>0)
		{
			if (m_recvDropPackets>=s_dropThresholdForResend)
			{
				RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, m_recvDropPackets="<<m_recvDropPackets);
				if (m_recvDropPackets>s_dropThresholdForTick)
				{
					clearRecvDropTick = FALSE;
					if (++m_recvDropTick>=s_maxDropTick)
					{
						RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, m_recvDropTick="<<m_recvDropTick);
						callObserver = TRUE;
						m_recvDropTick = 0;
					}
				}
			}
			m_recvDropPackets = 0;
		}
		if (clearRecvDropTick)
			m_recvDropTick = 0;

		if (m_sendDropPackets>0)
		{
			if (m_sendDropPackets>=s_dropThresholdForResend)
			{
				if (m_resendIncrement<RUDP_MAX_RESEND_INCREMENT)
				{
					RUDP_INFO_TRACE_THIS("Add resend time, inc="<<m_resendIncrement<<" m_sendDropPackets="<<m_sendDropPackets);
					++m_resendIncrement;
				}
				else
				{
					RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, m_sendDropPackets="<<m_sendDropPackets);

					if (m_sendDropPackets>s_dropThresholdForTick)
					{
						clearSendDropTick = FALSE;
						if (++m_sendDropTick>=s_maxDropTick)
						{
							RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, m_sendDropTick="<<m_sendDropTick);
							callObserver = TRUE;
							m_sendDropTick = 0;
						}
					}
				}
			}
			m_sendDropPackets = 0;
		}
		else if (m_resendIncrement>RUDP_MAX_RESEND_INCREMENT/2)
		{
			RUDP_INFO_TRACE_THIS("Reduce resend time, inc="<<m_resendIncrement);
			--m_resendIncrement;
		}
		if (clearSendDropTick)
			m_sendDropTick = 0;

		if (callObserver && m_observer)
			m_observer->OnObserve("RUDP");

		return;
	}

	if (m_status==RUDP_STATE_CLOSED)
	{
		// Try sending data again and then release.
		UpdateSending(GetRtRudpTimeUS());
		Release();
		ReleaseReference();
		return;
	}

	RUDP_WARNING_TRACE_THIS("CRtRudpConn::OnTimer, connection handshake timeout! state="<<m_status);
	OnDisconnect(RT_ERROR_NETWORK_CONNECT_TIMEOUT, m_lowerTransport.ParaIn());
	return;
}

RtResult CRtRudpConn::RunSendCycle(RtRudpTimeUS time)
{
// 	if (m_status==RUDP_STATE_CLOSED)
// 		return;

	if (time < lastUpdateTime)
	{
// 		RUDP_INFO_TRACE_THIS("CRtRudpConn::RunSendCycle, unexpected timestamp! time="<<time<<" lastUpdateTime="<<lastUpdateTime);
		// Always set the last time in case of overflow
		lastUpdateTime=time;
// 		return RT_OK;
	}

	RtRudpTimeUS diff = time - lastAckTime;
	if ( !resendList.IsEmpty() && time > lastAckTime && lastAckTime && diff > MAX_PING_VALUE*PING_MULTIPLIER_TO_RESEND*1000 )
	{
		resendTimeIncrement = (diff<10000000) ? diff : 10000000;
/*
		if (diff>3000000)
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::RunSendCycle,"
			" diff="<<diff/1000<<
			" inc="<<resendTimeIncrement/1000<<
			" ping="<<ping);
*/

		if (diff > timeoutTime)
		{
			// dead connection
			// We've waited a very long time for a reliable packet to get an ack and it never has
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::RunSendCycle, connection lost! Long time no ack!");
			OnDisconnect(RT_ERROR_NETWORK_CONNECTION_LOST, m_lowerTransport.ParaIn());
			return RT_ERROR_NETWORK_CONNECTION_LOST;
		}
	}

#ifndef RUDP_SEND_IMMEDIATELY
 	WrapUpPdusToPackets(/*time*/);
#endif

	RtResult rv = UpdateSending(time);
	if (RT_FAILED(rv))
	{
#ifdef RT_WIN32
		// Under Linux platform, dropping packet via iptable or stopping network service will make 
		// sending udp data failed. In this case, we'd like the connection to keep going. 
		if (rv!=RT_ERROR_PARTIAL_DATA)
		{
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::RunSendCycle, UpdateSending failed! rv="<<rv);
			OnDisconnect(rv, m_lowerTransport.ParaIn());
		}
#endif
		return rv;
	}

/*
	if (time - m_lastWarningResendQueueTime > 60*1000000 && resendQueue.Size()>=500)
	{
		m_lastWarningResendQueueTime = time;
		RUDP_WARNING_TRACE_THIS("CRtRudpConn::RunSendCycle, resend queue too long!"
			" size="<<resendQueue.Size()<<
			" inc="<<resendTimeIncrement/1000<<
			" ping="<<ping);
	}
*/
	
	/// CheckOnSend() moved here to cover the case that all buffered data has been sent out by network thread 
	/// before SendData() return failure, which results in user thread can not get OnSend() message any more.
	CheckOnSend();

	return RT_OK;
}

RtResult CRtRudpConn::UpdateSending(RtRudpTimeUS time, BOOL updateResendQueue)
{
// 	RtRudpTimeUS elapsedTime = time - lastUpdateTime;
	lastUpdateTime=time;
// 	availableBandwidth+=currentBandwidth * ((double)elapsedTime/1000000.0f);
// 	if (availableBandwidth > currentBandwidth)
		availableBandwidth = currentBandwidth;

	// unsigned resendListSize;
	int reliableDataSentNum = 0;

	// Water canister has to have enough room to put more water in :)
	double requiredBuffer=(float)(m_mtuSize*8);
	if (requiredBuffer > currentBandwidth)
		requiredBuffer=currentBandwidth;

	RtResult rv = RT_OK;
	if (updateBitStream.GetNumberOfBitsUsed()>0)
	{
		// Send data that was sent out failed last time.
		rv = SendBitStream(&updateBitStream);
		if (RT_FAILED(rv))
		{
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::UpdateSending 1, SendBitStream fail! rv="<<rv);
			return rv;
		}

		updateBitStream.Reset();
	}

	while ( availableBandwidth > requiredBuffer )
	{
		GenerateDatagram( &updateBitStream, reliableDataSentNum, time, updateResendQueue );
		if ( updateBitStream.GetNumberOfBitsUsed() > 0 )
		{
			availableBandwidth-=updateBitStream.GetNumberOfBitsUsed()+UDP_HEADER_SIZE*8;
			rv = SendBitStream( &updateBitStream );
			if (RT_FAILED(rv))
			{
				RUDP_WARNING_TRACE_THIS("CRtRudpConn::UpdateSending 2, SendBitStream fail! rv="<<rv);
				// Send data failed. Cache the updateBitStream and try later.
				if (rv!=RT_ERROR_PARTIAL_DATA)
				{
					return rv;
				}
				break;
			}
			updateBitStream.Reset();
		}
		else
			break;
	}

/*
	if (reliableDataSentNum>0)
		m_lastReliableSendTime = time;
*/

#ifdef RUDP_ENABLE_BW_DETECT
	bool lastContinuousSend=continuousSend;
	continuousSend=availableBandwidth < requiredBuffer;

	if (continuousSend==true && lastContinuousSend==false)
	{
		histogramAckCount=0;
		histogramPlossCount=0;
		histogramStartTime=time+ping*2*1000;
		histogramEndTime=histogramStartTime+MAX_TIME_TO_SAMPLE;
		++histogramReceiveMarker;
	}
	else if ( time >= histogramEndTime )
	{
		float packetloss;

		double delta;
		if (histogramAckCount+histogramPlossCount)
			packetloss=(float)histogramPlossCount / ((float)histogramAckCount+(float)histogramPlossCount);
		else
			packetloss=0.0f; // This line can be true if we are sending only acks

		if (continuousSend==false)
		{
			if (packetloss > PACKETLOSS_TOLERANCE)
			{
				highBandwidth=currentBandwidth;
				if (packetloss > .3)
				{
					lowBandwidth/=2;
				}
				else
				{
					lowBandwidth*=.9;
				}

				if (lowBandwidth < MINIMUM_SEND_BPS)
					lowBandwidth=MINIMUM_SEND_BPS;

				delta = (highBandwidth-lowBandwidth)/2;
				currentBandwidth=delta+lowBandwidth;
				noPacketlossIncreaseCount=0;
			}
		}
		else
		{
			if (packetloss <= PACKETLOSS_TOLERANCE)
				lowBandwidth=currentBandwidth;
			else
				highBandwidth=currentBandwidth;

			if (packetloss==0.0)
			{
				// If no packetloss for many increases in a row, drop the high range and go into search mode.
//				if (++noPacketlossIncreaseCount==10)
				if (++noPacketlossIncreaseCount==3)
				{
					noPacketlossIncreaseCount=0;
					highBandwidth=0;
				}
			}
			else
				noPacketlossIncreaseCount=0;

			if (highBandwidth!=0.0)
			{
				// If a lot of packetloss at any time, decrease the low range by half
				if (packetloss > .3)
				{
					lowBandwidth/=2;
					if (lowBandwidth < MINIMUM_SEND_BPS)
						lowBandwidth=MINIMUM_SEND_BPS;
				}

				delta = (highBandwidth-lowBandwidth)/2;
				if (delta < MINIMUM_SEND_BPS/4)
				{
					// If no packetloss and done searching, increase the high range by 50%
					if (packetloss==0.0)
					{
						highBandwidth*=1.5;
					}
					else if (packetloss < PACKETLOSS_TOLERANCE)
					{
						// If some packetloss and done searching, increase the high range by 5%
						highBandwidth*=1.2;
					}
					else if (packetloss < PACKETLOSS_TOLERANCE*2)
					{
						// If some packetloss, but not a huge amount and done searching, decrease the low range by 10%
						lowBandwidth*=.9;
						if (lowBandwidth < MINIMUM_SEND_BPS)
							lowBandwidth=MINIMUM_SEND_BPS;
					}
					delta = (highBandwidth-lowBandwidth)/2;
				}
				currentBandwidth=delta+lowBandwidth;
			}
			else
			{
				// Don't know the maximum bandwidth, so keep doubling to find out
				currentBandwidth*=2.0;
			}
		}

		histogramPlossCount=0;
		histogramAckCount=0;
		histogramStartTime=time+TIME_TO_NEW_SAMPLE+ping*2*1000;
		histogramEndTime=histogramStartTime+MAX_TIME_TO_SAMPLE;
		++histogramReceiveMarker;
	}
#endif // RUDP_ENABLE_BW_DETECT

#ifdef RUDP_ENABLE_STATISTICS
	if (currentBandwidth>m_statistics.m_maxBandwidth)
		m_statistics.m_maxBandwidth = currentBandwidth;
	if (currentBandwidth<m_statistics.m_minBandwidth)
		m_statistics.m_minBandwidth = currentBandwidth;
#endif

	return rv;
}

void CRtRudpConn::InitializeVariables( void )
{
	memset( waitingForOrderedPacketReadIndex, 0, NUMBER_OF_ORDERED_STREAMS * sizeof(RtRudpOrderingID));
	memset( waitingForSequencedPacketReadIndex, 0, NUMBER_OF_ORDERED_STREAMS * sizeof(RtRudpOrderingID) );
	memset( waitingForOrderedPacketWriteIndex, 0, NUMBER_OF_ORDERED_STREAMS * sizeof(RtRudpOrderingID) );
	memset( waitingForSequencedPacketWriteIndex, 0, NUMBER_OF_ORDERED_STREAMS * sizeof(RtRudpOrderingID) );
	splitPacketId = 0;
	messageNumber = 0;
	m_lastPacketReceiveTime = /*m_lastReliableSendTime = */lastUpdateTime = GetRtRudpTimeUS();

	lastAckTime = 0;

	hasReceivedPacketQueue.ClearAndForceAllocation(DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE);
	receivedPacketsBaseIndex=0;

	SetPing( 100 );
	
//	resendList.Preallocate(RESEND_TREE_ORDER);

	currentBandwidth=STARTING_SEND_BPS;
	availableBandwidth=0;
#ifdef RUDP_ENABLE_BW_DETECT
	highBandwidth=lowBandwidth=STARTING_SEND_BPS;
	histogramStartTime=lastUpdateTime+ping*2*1000;
	histogramEndTime=histogramStartTime+MAX_TIME_TO_SAMPLE;
	histogramPlossCount=0;
	histogramAckCount=0;
	continuousSend=false;
	histogramReceiveMarker=0;
	noPacketlossIncreaseCount=0;
#endif // RUDP_ENABLE_BW_DETECT

#ifdef RUDP_ENABLE_STATISTICS
	m_statistics.m_connectionStartTime = GetTickCountEx();
	m_statistics.m_minBandwidth = STARTING_SEND_BPS * 2;
#endif
}

inline void CRtRudpConn::SetPing( RtRudpTime i )
{
	ping = (i>MAX_PING_VALUE) ? MAX_PING_VALUE : (i<30 ? 30 : i);
 	resendTimeIncrement = (ping<80) ? ping*3000 : ( ping<120 ? 240000 : ping*2000);
//	resendTimeIncrement = (ping<80) ? 240000 : ping*3000;
}

/// CRtRudpConn::SendBuffered() should only be called in user thread.
RtResult CRtRudpConn::SendBuffered(
		RtRudpPduType aType,
		CRtMessageBlock &aData, 
		RtRudpReliability aReliability,
		DataPriority aPriority,
		RudpChannel aChannelID,
		INT aTTL,
		BOOL inDestroyData)
{
	if (m_status!=RUDP_STATE_OPEN)
		return RT_ERROR_NOT_AVAILABLE;

	DWORD dataLen = aData.GetChainedLength();
	if (dataLen==0 && aType==RUDP_PDU_DATA)
	{
		RUDP_ERROR_TRACE_THIS("CRtRudpConn::SendBuffered, dataLen should not be 0!");
		return RT_OK;
	}

	if (aType==RUDP_PDU_DATA)
	{
		if (RUDP_SEND_BUFFER_SIZE>=m_maxSendBufferSize || m_needOnSend)
		{
			m_needOnSend = TRUE;
			return RT_ERROR_PARTIAL_DATA;
		}
		m_totalSizeOfPassedInData += dataLen;
	}

	// Fix any bad parameters
	if ( aReliability > RUDP_RELIABLE_ORDERED || aReliability < RUDP_UNRELIABLE )
		aReliability = RUDP_RELIABLE_ORDERED;

	if ( aPriority > IRtTransport::DP_LOW || aPriority < IRtTransport::DP_SYSTEM )
		aPriority = IRtTransport::DP_MEDIUM;

	if ( aChannelID >= NUMBER_OF_ORDERED_STREAMS )
		aChannelID = 0;

	RtRudpPdu* pdu = m_sendPduBuffer.WriteLock();
	
	pdu->pduType = aType;
	pdu->data = inDestroyData ? &aData : aData.DuplicateChained();
	pdu->reliability = aReliability;
	pdu->priority = aPriority;
	pdu->orderingChannel = aChannelID;
	pdu->ttl = (aTTL!=-1) ? (aTTL+m_resendIncrement) : -1;

	m_sendPduBuffer.WriteUnlock();

#ifdef RUDP_SEND_IMMEDIATELY
	if (CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()))
	{
		SendBuffered_i();
	}
	else
	{
		int pduBufferSize = m_sendPduBuffer.Size();
		if (pduBufferSize<2 || pduBufferSize>100)
		{
			if (pduBufferSize>100)
			{
				/// Check bug. 
				RUDP_ERROR_TRACE_THIS("CRtRudpConn::SendBuffered, pduBufferSize="<<pduBufferSize);
			}

			CRtRudpSendBufferedEvent* pEvent = new CRtRudpSendBufferedEvent(this);
			m_pThreadNetwork->GetEventQueue()->PostEvent(pEvent);
		}
	}
#endif // RUDP_SEND_IMMEDIATELY

#ifdef RUDP_FLOWCONTROL_OPTIMIZE
	if (!m_onFlowControl)
	{
		if (CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()))
			InsertIntoFlowControl();
		else 
		{
			CRtRudpInsertIntoFlowControlEvent* pEvent = new CRtRudpInsertIntoFlowControlEvent(this);
			m_pThreadNetwork->GetEventQueue()->PostEvent(pEvent);
		}
	}
#endif // RUDP_FLOWCONTROL_OPTIMIZE
	return RT_OK;
}

RtResult CRtRudpConn::SendPdu(
		  RtRudpPduType aType,
		  CRtMessageBlock &aData, 
		  RtRudpReliability aReliability,
		  DataPriority aPriority,
		  RudpChannel aChannelID,
		  INT aTTL,
		  BOOL aDuplicateData)
{
	if (m_status==RUDP_STATE_CLOSED)
		return RT_ERROR_NETWORK_SOCKET_CLOSE;

	int numberOfBytesToSend=aData.GetChainedLength();
/*
	if ( numberOfBytesToSend == 0 )
		return RT_OK;
*/

	RtRudpPacket * internalPacket = m_pFlowControl->m_internalPacketPool->New();

	internalPacket->data = aDuplicateData ? aData.DuplicateChained() : &aData;
	internalPacket->lastSendTime = 0;

	internalPacket->pduType = aType;
	internalPacket->priority = aPriority;
	internalPacket->reliability = aReliability;
	internalPacket->splitPacketCount = 0;
	internalPacket->ttl = aTTL;

	// Calculate if I need to split the packet
	int headerLength = BITS_TO_BYTES( GetBitStreamHeaderLength( internalPacket ) );

	int maxDataSize = m_mtuSize - UDP_HEADER_SIZE - headerLength;

	bool splitPacket = numberOfBytesToSend > maxDataSize;

	if ( internalPacket->reliability == RUDP_RELIABLE_SEQUENCED || internalPacket->reliability == RUDP_UNRELIABLE_SEQUENCED )
	{
		// Assign the sequence stream and index
		internalPacket->orderingChannel = aChannelID;
		internalPacket->orderingIndex = waitingForSequencedPacketWriteIndex[ aChannelID ] ++;
	}
	else if ( internalPacket->reliability == RUDP_RELIABLE_ORDERED )
	{
		// Assign the ordering channel and index
		internalPacket->orderingChannel = aChannelID;
		internalPacket->orderingIndex = waitingForOrderedPacketWriteIndex[ aChannelID ] ++;
	}

	if ( splitPacket )   // If it uses a secure header it will be generated here
	{
#ifdef RUDP_ENABLE_STATISTICS
		++m_statistics.m_numberOfSplitPacket;
#endif

		// Must split the packet.  This will also generate the SHA1 if it is required. It also adds it to the send list.
		SplitPacket( internalPacket );
	}
	else
	{
#ifdef RUDP_ENABLE_STATISTICS
		++m_statistics.m_numberOfUnsplitPacket;
#endif

		sendPacketSet[ internalPacket->priority ].Push( internalPacket );
	}

	return RT_OK;
}

void CRtRudpConn::SplitPacket( RtRudpPacket *internalPacket )
{
	// Doing all sizes in bytes in this function so I don't write partial bytes with split packets
	internalPacket->splitPacketCount = 1; // This causes GetBitStreamHeaderLength to account for the split packet header
	int headerLength = BITS_TO_BYTES( GetBitStreamHeaderLength( internalPacket ) );
	int dataByteLength = internalPacket->data->GetChainedLength();
	int maximumSendBlock, bytesToSend;
	RtRudpSplitPacketID splitPacketIndex;
	RtRudpPacket **internalPacketArray;

	// How much to send in the largest block
	maximumSendBlock = m_mtuSize - UDP_HEADER_SIZE - headerLength;

	// Calculate how many packets we need to create
	internalPacket->splitPacketCount = ( ( dataByteLength - 1 ) / ( maximumSendBlock ) + 1 );
	RT_ASSERTE(internalPacket->splitPacketCount>=2);

#ifdef RUDP_ENABLE_STATISTICS
	m_statistics.m_totalSplits += internalPacket->splitPacketCount;
#endif

	internalPacket->splitPacketId = splitPacketId++;
	internalPacket->splitPacketIndex = 0;

	// Optimization
	bool usedAlloca=false;
	if (sizeof( RtRudpPacket* ) * internalPacket->splitPacketCount < MAX_ALLOCA_STACK_ALLOCATION)
	{
		internalPacketArray = ( RtRudpPacket** ) alloca( sizeof( RtRudpPacket* ) * (internalPacket->splitPacketCount) );
		usedAlloca=true;
	}
	else
		internalPacketArray = new RtRudpPacket*[internalPacket->splitPacketCount];

	internalPacketArray[0] = internalPacket;
	sendPacketSet[ internalPacket->priority ].Push( internalPacket );

	splitPacketIndex = 1;
	CRtMessageBlock* pMb = internalPacket->data;
	do
	{
		internalPacketArray[ splitPacketIndex ] = m_pFlowControl->m_internalPacketPool->New();
		memcpy( internalPacketArray[ splitPacketIndex ], internalPacket, sizeof( RtRudpPacket ) );

		bytesToSend = dataByteLength - (splitPacketIndex-1) * maximumSendBlock;
		if ( bytesToSend > maximumSendBlock )
			bytesToSend = maximumSendBlock;

		pMb = pMb->Disjoint(bytesToSend);
		internalPacketArray[ splitPacketIndex ]->data = pMb;
		internalPacketArray[ splitPacketIndex ]->splitPacketIndex = splitPacketIndex;

		sendPacketSet[ internalPacket->priority ].Push( internalPacketArray[ splitPacketIndex ] );
	}
	while ( ++splitPacketIndex < internalPacket->splitPacketCount );

	if (usedAlloca==false)
		delete [] internalPacketArray;
}

unsigned CRtRudpConn::GenerateDatagram( RakNet::BitStream *output, int& reliableDataSentNum, RtRudpTimeUS time, BOOL updateResendQueue )
{
	RtRudpPacket * internalPacket;
	int maxDataBitSize;
	int nextPacketBitLength;
	unsigned i;
	bool writeFalseToHeader = true;
	unsigned messagesSent=0;

	maxDataBitSize = (m_mtuSize - UDP_HEADER_SIZE)<<3;

	if (acknowlegements.Size()>0)
	{
		/// To ensure that the frequency of ACK packets sent separately will not be too frequent.
		if (RUDP_SEND_BUFFER_SIZE>0 || !m_isAckSent)
		{
			m_isAckSent = TRUE;
			output->Write(true);
			writeFalseToHeader=false;
			++messagesSent;
			acknowlegements.Serialize(output, (m_mtuSize-UDP_HEADER_SIZE)*8-1, true);
		}
		else
			m_isAckSent = FALSE;
	}

	if (updateResendQueue)
	{
	while ( !resendQueue.IsEmpty() )
	{
		internalPacket = resendQueue.Peek();
		// The resend Queue can have holes.  This is so we can deallocate blocks without having to compress the array
		if ( internalPacket->lastSendTime == 0 )
		{
			resendQueue.Pop();
//			internalPacket->data->DestroyChained();
			m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
			continue; // This was a hole
		}

		if ( (internalPacket->lastSendTime+resendTimeIncrement) < time )
		{
			// ttl in milliseconds. 
			//if (internalPacket->pduType!=RUDP_PDU_DROPPACKET && internalPacket->ttl!=-1 && internalPacket->creationTime+internalPacket->ttl*1000<time)

			// ttl in RTTs. 
 			//if (internalPacket->pduType!=RUDP_PDU_DROPPACKET && internalPacket->ttl!=-1 && internalPacket->creationTime+ping*internalPacket->ttl*1000<time)

			// ttl in resending times. 
			if (internalPacket->pduType!=RUDP_PDU_DROPPACKET && internalPacket->ttl!=-1 && (internalPacket->ttl--)<=0 )
			{
				/// TTL timeout!
				++m_sendDropPackets;
				if (internalPacket->data)
				{
					DecreaseSendBufferSize(*internalPacket);
					internalPacket->data->DestroyChained();
					internalPacket->data = NULL;
				}

#if 1
				/// Just give up sending the data pdu.
				resendList.Delete(internalPacket->packetNo);
				resendQueue.Pop();
				m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
				continue; 
#else
				internalPacket->pduType = RUDP_PDU_DROPPACKET;
#endif

#ifdef RUDP_ENABLE_STATISTICS
				++m_statistics.m_dropPacketsSent;
#endif
			}

			nextPacketBitLength = GetBitStreamHeaderLength( internalPacket ) + BYTES_TO_BITS(internalPacket->data ? internalPacket->data->GetChainedLength() : 0);

			if ( output->GetNumberOfBitsUsed() + nextPacketBitLength > maxDataBitSize )
			{
				// Not enough room to use this packet after all!
				goto END_OF_GENERATE_FRAME;
			}

			internalPacket = resendQueue.Pop();

			// Write to the output bitstream

			if (writeFalseToHeader)
			{
				output->Write(false);
				writeFalseToHeader=false;
			}

#ifdef RUDP_ENABLE_STATISTICS
			++m_statistics.m_packetsResent;
			m_statistics.m_totalBitsResent += 
#endif
				WriteToBitStreamFromInternalPacket( output, internalPacket );

			++messagesSent;
			++reliableDataSentNum;

			internalPacket->lastSendTime = time;
#ifdef RUDP_ENABLE_BW_DETECT
			if (time >= histogramStartTime && internalPacket->histogramMarker==histogramReceiveMarker)
				histogramPlossCount++;

			internalPacket->histogramMarker=histogramReceiveMarker;
#endif // RUDP_ENABLE_BW_DETECT

			// Put the packet back into the resend list at the correct spot
			// Don't make a copy since I'm reinserting an allocated struct
			InsertPacketIntoResendList( internalPacket, time, false );
		}
		else
		{
			break;
		}
	}
	} // if (updateResendQueue)

	// From highest to lowest priority, fill up the output bitstream from the send lists
	for ( i = 0; i < DP_NUMBER_OF_PRIORITIES; i++ )
	{
		while ( !sendPacketSet[ i ].IsEmpty() )
		{
			internalPacket = sendPacketSet[ i ].Pop();

			nextPacketBitLength = GetBitStreamHeaderLength( internalPacket ) + BYTES_TO_BITS(internalPacket->data->GetChainedLength());

			if ( output->GetNumberOfBitsUsed() + nextPacketBitLength > maxDataBitSize )
			{
				// This output won't fit.
				sendPacketSet[ i ].PushAtHead( internalPacket ); // Push this back at the head so it is the next thing to go out
				break;
			}

			// Write to the output bitstream

			if (writeFalseToHeader)
			{
				output->Write(false);
				writeFalseToHeader=false;
			}

			bool isReliable = false;
			if ( RUDP_IS_RELIABLE(internalPacket->reliability) )
			{
				isReliable = true;
				// Only reliable pdu need packet number.
				internalPacket->packetNo = messageNumber++;

#ifdef RUDP_ENABLE_CHECK_PACKETNO
				static char s_sendPacketNo[64] = "";
				if (internalPacket->reliability==RUDP_RELIABLE_ORDERED)
					snprintf(s_sendPacketNo, 63, "%u(%u) ", internalPacket->packetNo, internalPacket->orderingIndex);
				else
					snprintf(s_sendPacketNo, 63, "%u ", internalPacket->packetNo);
				s_sendPacketNoList.append(s_sendPacketNo);
				if (s_sendPacketNoList.length()>1900)
				{
					RUDP_INFO_TRACE_THIS("SEND PacketNo List: "<<s_sendPacketNoList);
					s_sendPacketNoList = "";
				}
#endif // RUDP_ENABLE_CHECK_PACKETNO
			}

#ifdef RUDP_ENABLE_STATISTICS
			++m_statistics.m_packetsSent;
			m_statistics.m_packetDataBitsSent += (internalPacket->data ? (internalPacket->data->GetChainedLength() << 3) : 0);
			m_statistics.m_packetTotalBitsSent += 
#endif
				WriteToBitStreamFromInternalPacket( output, internalPacket );
			++messagesSent;

			if ( isReliable )
			{
#ifdef RUDP_ENABLE_STATISTICS
				++m_statistics.m_reliablePacketsSent;
#endif

				// Reliable packets are saved to resend later

				internalPacket->creationTime = time;
				internalPacket->lastSendTime = time;
#ifdef RUDP_ENABLE_BW_DETECT
				internalPacket->histogramMarker=histogramReceiveMarker;
#endif
				resendList.Insert( internalPacket->packetNo, internalPacket);
				InsertPacketIntoResendList( internalPacket, time, true);
				++reliableDataSentNum;
			}
			else
			{
				// Unreliable packets are deleted
				DecreaseSendBufferSize(*internalPacket);
//				internalPacket->data->DestroyChained();
				m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
			}
		}
	}

END_OF_GENERATE_FRAME:

	return messagesSent;
}

int CRtRudpConn::GetBitStreamHeaderLength( const RtRudpPacket *const internalPacket )
{
	// Write the pdu type (1 byte)
	//bitStream->Write( internalPacket->pduType );
	int bitLength = sizeof(RtRudpPduType)*8;

	// Write the PacketReliability.  This is encoded in 3 bits
	//bitStream->WriteBits((unsigned char*)&(internalPacket->reliability), 3, true);
	bitLength += 3;

	// If the reliability requires an ordering channel and ordering index, we Write those.
	if ( internalPacket->reliability == RUDP_UNRELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_ORDERED )
	{
		// ordering channel encoded in 5 bits (from 0 to 31)
		//bitStream->WriteBits((unsigned char*)&(internalPacket->orderingChannel), 5, true);
		bitLength+=5;

		// ordering index is one byte
		//bitStream->WriteCompressed(internalPacket->orderingIndex);
		bitLength+=sizeof(RtRudpOrderingID)*8;
	}

	if ( RUDP_IS_RELIABLE(internalPacket->reliability) )
	{
		// Write the message number (2 bytes)
		//bitStream->Write( internalPacket->packetNo );
		bitLength+=sizeof(RtRudpPacketNo)*8;
	}

	// Write if this is a split packet (1 bit)
	bool isSplitPacket = internalPacket->splitPacketCount > 0;

	//bitStream->Write(isSplitPacket);
	bitLength += 1;

	if ( isSplitPacket )
	{
		// split packet indices are two bytes (so one packet can be split up to 65535
		// times - maximum packet size would be about 500 * 65535)
		//bitStream->Write(internalPacket->splitPacketId);
		//bitStream->WriteCompressed(internalPacket->splitPacketIndex);
		//bitStream->WriteCompressed(internalPacket->splitPacketCount);
		bitLength += (sizeof(RtRudpSplitPacketNo) + sizeof(RtRudpSplitPacketID) * 2) * 8 + 2;
	}

	// Read how many bytes the packet data is.  Stored in 16 bits(Webb: plus 1 bit as WriteCompressed() might add one bit in length.)
	bitLength += 17;

	return bitLength;
}

int CRtRudpConn::WriteToBitStreamFromInternalPacket( RakNet::BitStream *bitStream, const RtRudpPacket *const internalPacket )
{
	int start = bitStream->GetNumberOfBitsUsed();
	const unsigned char c = (unsigned char) internalPacket->reliability;

	// Write the pdu type (1 byte)
	bitStream->Write( internalPacket->pduType );

	// Acknowledgment packets have no more data than the messageNumber and whether it is anacknowledgment

	// Write the PacketReliability.  This is encoded in 3 bits
	bitStream->WriteBits( (const unsigned char *)&c, 3, true );

	// If the reliability requires an ordering channel and ordering index, we Write those.
	if ( internalPacket->reliability == RUDP_UNRELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_ORDERED )
	{
		// ordering channel encoded in 5 bits (from 0 to 31)
		bitStream->WriteBits( ( unsigned char* ) & ( internalPacket->orderingChannel ), 5, true );

		// One or two bytes
		bitStream->Write( internalPacket->orderingIndex );
	}

	if ( RUDP_IS_RELIABLE(internalPacket->reliability) )
	{
		// Write the message number (2 bytes)
		bitStream->Write( internalPacket->packetNo );
	}

	// Write if this is a split packet (1 bit)
	bool isSplitPacket = internalPacket->splitPacketCount > 0;

	bitStream->Write( isSplitPacket );

	if ( isSplitPacket )
	{
		bitStream->Write( internalPacket->splitPacketId );
		bitStream->WriteCompressed( internalPacket->splitPacketIndex );
		bitStream->WriteCompressed( internalPacket->splitPacketCount );
	}

	// Write how many bits the packet data is. Stored in 13 bits
	// Never send more than MTU_SIZE bytes

	unsigned short byteLength = 0;
	if (internalPacket->data)
	{
		byteLength = ( unsigned short ) internalPacket->data->GetChainedLength(); // Ignore the 2 high bytes for WriteBits
	}

	bitStream->WriteCompressed(byteLength);

	if (byteLength>0)
	{
		// Write the actual data.
#if 1
		bitStream->WriteAlignedBytes( (unsigned char*)internalPacket->data->GetTopLevelReadPtr(), internalPacket->data->GetTopLevelLength() );
		CRtMessageBlock* pNext = internalPacket->data->GetNext();
		while (pNext)
		{
			bitStream->Write( pNext->GetTopLevelReadPtr(), pNext->GetTopLevelLength() );
			pNext = pNext->GetNext();
		}
#else
		bitStream->WriteAlignedBytes( ( unsigned char* ) ( (byteLength==internalPacket->data->GetTopLevelLength()) ? 
			internalPacket->data->GetTopLevelReadPtr() : internalPacket->data->FlattenChained().c_str() ), byteLength );
#endif
	}

	return bitStream->GetNumberOfBitsUsed() - start;
}

void CRtRudpConn::InsertPacketIntoResendList( RtRudpPacket *internalPacket, RtRudpTimeUS time, bool firstResend )
{
	// lastAckTime is the time we last got an acknowledgment - however we also initialize the value if this is the first resend and
	// either we never got an ack before or we are inserting into an empty resend queue
	if ( firstResend && (lastAckTime == 0 || resendList.IsEmpty()))
	{
		lastAckTime = time; // Start the timer for the ack of this packet if we aren't already waiting for an ack
	}

	resendQueue.Push( internalPacket );
}

inline RtResult CRtRudpConn::SendBitStream( RakNet::BitStream *bitStream )
{
	int len = bitStream->GetNumberOfBytesUsed();

#ifdef RUDP_ENABLE_STATISTICS
	m_statistics.m_totalBytesSent += len;
	++m_statistics.m_packagesSent;
#endif

	return SendData_i(len, bitStream->GetData());
}

unsigned CRtRudpConn::RemovePacketFromResendList( const RtRudpPacketNo packetNo, RtRudpTimeUS time, RtRudpTime& rtt )
{
	RtRudpPacket* internalPacket;
	bool deleted = resendList.Delete(packetNo, internalPacket);
	if (deleted)
	{
#ifdef RUDP_ENABLE_STATISTICS
		++m_statistics.m_reliablePacketsAcked;
#endif

		rtt = (time - internalPacket->lastSendTime)/1000;
		internalPacket->lastSendTime=0; // Will be freed in the update function
		if (internalPacket->data)
		{
			DecreaseSendBufferSize(*internalPacket);
			internalPacket->data->DestroyChained();
			internalPacket->data = NULL;
		}
		return internalPacket->histogramMarker;
	}

	rtt = 0;
	return (unsigned)-1;
}

RtRudpPacket* CRtRudpConn::CreateInternalPacketFromBitStream( RakNet::BitStream *bitStream, RtRudpTimeUS time )
{
	bool bitStreamSucceeded;
	RtRudpPacket* internalPacket;

	if ( bitStream->GetNumberOfUnreadBits() < (int) sizeof( internalPacket->pduType ) * 8 )
		return 0; // leftover bits

	internalPacket = m_pFlowControl->m_internalPacketPool->New();

	internalPacket->creationTime = time;

	CRtString errpos;
	{
		// Read the pdu type
		bitStreamSucceeded = bitStream->Read(internalPacket->pduType);
		if ( bitStreamSucceeded == false )
		{
			errpos = "pduType";
			goto CreateInternalPacketFail;
		}

		// Read the PacketReliability. This is encoded in 3 bits
		unsigned char reliability;
		bitStreamSucceeded = bitStream->ReadBits( ( unsigned char* ) ( &( reliability ) ), 3 );
		internalPacket->reliability = ( RtRudpReliability ) reliability;
		if ( bitStreamSucceeded == false )
		{
			errpos = "reliability";
			goto CreateInternalPacketFail;
		}

		// If the reliability requires an ordering channel and ordering index, we read those.
		if ( internalPacket->reliability == RUDP_UNRELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_SEQUENCED || internalPacket->reliability == RUDP_RELIABLE_ORDERED )
		{
			// ordering channel encoded in 5 bits (from 0 to 31)
			bitStreamSucceeded = bitStream->ReadBits( ( unsigned char* ) & ( internalPacket->orderingChannel ), 5 );
			if ( bitStreamSucceeded == false )
			{
				errpos = "channelID";
				goto CreateInternalPacketFail;
			}

			bitStreamSucceeded = bitStream->Read( internalPacket->orderingIndex );
			if ( bitStreamSucceeded == false )
			{
				errpos = "orderingID";
				goto CreateInternalPacketFail;
			}
		}

		if ( RUDP_IS_RELIABLE(internalPacket->reliability) )
		{
			// Read the packet number (2 bytes)
			bitStreamSucceeded = bitStream->Read( internalPacket->packetNo );
			if ( bitStreamSucceeded == false )
			{
				errpos = "packetNo";
				goto CreateInternalPacketFail;
			}
		}

		// Read if this is a split packet (1 bit)
		bool isSplitPacket;
		bitStreamSucceeded = bitStream->Read( isSplitPacket );
		if ( bitStreamSucceeded == false )
		{
			errpos = "isSplitPacket";
			goto CreateInternalPacketFail;
		}

		if ( isSplitPacket )
		{
			bitStreamSucceeded = bitStream->Read( internalPacket->splitPacketId );
			if ( bitStreamSucceeded == false )
			{
				errpos = "splitPacketID";
				goto CreateInternalPacketFail;
			}

			bitStreamSucceeded = bitStream->ReadCompressed( internalPacket->splitPacketIndex );
			if ( bitStreamSucceeded == false )
			{
				errpos = "splitPacketIndex";
				goto CreateInternalPacketFail;
			}

			bitStreamSucceeded = bitStream->ReadCompressed( internalPacket->splitPacketCount );
			if ( bitStreamSucceeded == false )
			{
				errpos = "splitPacketCount";
				goto CreateInternalPacketFail;
			}
		}
		else
			internalPacket->splitPacketIndex = internalPacket->splitPacketCount = 0;

		unsigned short byteLength;
		bitStreamSucceeded = bitStream->ReadCompressed( byteLength );
		if ( bitStreamSucceeded == false || byteLength >= MAXIMUM_MTU_SIZE )
		{
			errpos = "byteLength";
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::CreateInternalPacketFromBitStream, rv="<<bitStreamSucceeded<<" byteLength="<<byteLength);
			goto CreateInternalPacketFail;
		}

		if (byteLength>0)
		{
			unsigned char data[MAXIMUM_MTU_SIZE] = {0};
			// Read the data the packet holds
			bitStreamSucceeded = bitStream->ReadAlignedBytes( data, byteLength );
			if ( bitStreamSucceeded == false )
			{
				errpos = "data";
				RUDP_WARNING_TRACE_THIS("CRtRudpConn::CreateInternalPacketFromBitStream, byteLength="<<byteLength);
				goto CreateInternalPacketFail;
			}

			CRtMessageBlock mb(byteLength, (LPCSTR)data, CRtMessageBlock::DONT_DELETE, byteLength);
			internalPacket->data = mb.DuplicateChained();
		}
	}

	return internalPacket;

CreateInternalPacketFail:
	RUDP_WARNING_TRACE_THIS("CRtRudpConn::CreateInternalPacketFromBitStream, read "<<errpos
		<<" failed! pduType="<<internalPacket->pduType
		<<" packetNo="<<internalPacket->packetNo
		<<" unread="<<BITS_TO_BYTES(bitStream->GetNumberOfUnreadBits() )
		<<" total="<<bitStream->GetNumberOfBytesUsed() );
	BYTE* pData = bitStream->GetData();
	char dumpByte[16] = "";
	CRtString dumpData;
	for (WORD idx=0; idx<1024 && idx<bitStream->GetNumberOfBytesUsed(); ++idx)
	{
		sprintf(dumpByte, "%x ", pData[idx]);
		dumpData.append(dumpByte);
	}
	RUDP_WARNING_TRACE_THIS("CRtRudpConn::CreateInternalPacketFromBitStream, bitStream: "<<dumpData);

	m_pFlowControl->m_internalPacketPool->Delete( internalPacket );

	return 0;
}

void CRtRudpConn::DeleteOldUnreliableSplitPackets( RtRudpTimeUS time )
{
	unsigned i,j;
	i=0;
	while (i < splitPacketChannelList.Size())
	{
		if (time > splitPacketChannelList[i]->lastUpdateTime + 10000000 &&
			(splitPacketChannelList[i]->splitPacketList[0]->reliability==RUDP_UNRELIABLE || splitPacketChannelList[i]->splitPacketList[0]->reliability==RUDP_UNRELIABLE_SEQUENCED))
		{
			for (j=0; j < splitPacketChannelList[i]->splitPacketList.Size(); j++)
			{
//				splitPacketChannelList[i]->splitPacketList[j]->data->DestroyChained();
				m_pFlowControl->m_internalPacketPool->Delete(splitPacketChannelList[i]->splitPacketList[j]);
			}
			delete splitPacketChannelList[i];
			splitPacketChannelList.RemoveAtIndex(i);
		}
		else
			i++;
	}
}

bool CRtRudpConn::IsOlderOrderedPacket( RtRudpOrderingID newPacketOrderingIndex, RtRudpOrderingID waitingForPacketOrderingIndex )
{
	return newPacketOrderingIndex!=waitingForPacketOrderingIndex 
		&& RUDP_NOT_MORE_THAN(newPacketOrderingIndex, waitingForPacketOrderingIndex, RtRudpOrderingID);
}

void CRtRudpConn::InsertIntoSplitPacketList( RtRudpPacket * internalPacket, RtRudpTimeUS time )
{
	bool objectExists;
	unsigned index;
	index=splitPacketChannelList.GetIndexFromKey(internalPacket->splitPacketId, &objectExists);
	if (objectExists==false)
	{
		SplitPacketChannel *newChannel = new SplitPacketChannel;
		index=splitPacketChannelList.Insert(internalPacket->splitPacketId, newChannel);
	}
	splitPacketChannelList[index]->splitPacketList.Insert(internalPacket->splitPacketIndex, internalPacket);
	splitPacketChannelList[index]->lastUpdateTime=time;
}

RtRudpPacket * CRtRudpConn::BuildPacketFromSplitPacketList( RtRudpSplitPacketNo splitPacketId, RtRudpTimeUS time )
{
	unsigned i, j;
	bool objectExists;

	i=splitPacketChannelList.GetIndexFromKey(splitPacketId, &objectExists);
	RT_ASSERTE(objectExists);

	if (splitPacketChannelList[i]->splitPacketList.Size()==splitPacketChannelList[i]->splitPacketList[0]->splitPacketCount)
	{
		// Reconstruct
		RtRudpPacket * internalPacket = splitPacketChannelList[i]->splitPacketList[0];
		internalPacket->splitPacketCount = 0;
		internalPacket->creationTime = time;
		internalPacket->lastSendTime = 0;
		for (j=1; j < splitPacketChannelList[i]->splitPacketList.Size(); j++) {
			internalPacket->data->Append(splitPacketChannelList[i]->splitPacketList[j]->data);
			splitPacketChannelList[i]->splitPacketList[j]->data = NULL;
			m_pFlowControl->m_internalPacketPool->Delete(splitPacketChannelList[i]->splitPacketList[j]);
		}

		delete splitPacketChannelList[i];
		splitPacketChannelList.RemoveAtIndex(i);

		return internalPacket;
	}

	return 0;
}

void CRtRudpConn::AddToOrderingList( RtRudpPacket * internalPacket )
{
	if ( internalPacket->orderingChannel >= NUMBER_OF_ORDERED_STREAMS )
	{
		return;
	}

	if ( internalPacket->orderingChannel >= orderingList.Size() || orderingList[ internalPacket->orderingChannel ] == 0 )
	{
		// Need a linked list in this index
		orderingList.Replace( new RudpRecvOrderingListType, 0, internalPacket->orderingChannel );
	}

	RudpRecvOrderingListType *theList=orderingList[ internalPacket->orderingChannel ];
	RT_ASSERTE(theList);

	theList->End();
	theList->Add(internalPacket);
}

void CRtRudpConn::ClearSendPduBuffer()
{
	RtRudpPdu *pdu;

	while ((pdu=m_sendPduBuffer.ReadLock())!=0)
	{
		if (pdu->data) {
			pdu->data->DestroyChained();
			pdu->data = NULL;
		}

		m_sendPduBuffer.ReadUnlock();
	}
	m_sendPduBuffer.Clear();
}

#ifdef RUDP_FLOWCONTROL_OPTIMIZE

inline BOOL CRtRudpConn::InsertIntoFlowControl()
{
	if (!m_onFlowControl && NeedFlowControl())
	{
		m_onFlowControl = TRUE;
		m_pFlowControl->InsertConn(this);
		return TRUE;
	}
	return FALSE;
}

inline BOOL CRtRudpConn::RemoveFromFlowControl()
{
	if (m_onFlowControl && !NeedFlowControl())
	{
		m_onFlowControl = FALSE;
		if (!NeedFlowControl()) // This step is necessary to make sure thread safety.
		{
			m_pFlowControl->RemoveConn(this);
			return TRUE;
		}

		m_onFlowControl = TRUE;
	}
	return FALSE;
}

#endif // RUDP_FLOWCONTROL_OPTIMIZE

void CRtRudpConn::Close_i()
{
	m_timer.Cancel();

	/// !Move the RemoveConn() call to CRtRudpFlowControl::OnTimeout() to fix a crash bug. 
	///m_pFlowControl->RemoveConn(this);

	if(m_lowerTransport)
	{
		m_lowerTransport->Disconnect(RT_OK);
//		m_lowerTransport = NULL; 
	}
	m_sink = NULL;
}

void CRtRudpConn::Release()
{
	RT_ASSERTE(m_status==RUDP_STATE_CLOSED);

	if (m_pFlowControl==NULL)
		return;

	//////////////////////////////////////////////////////////////////////////
/*
	RUDP_INFO_TRACE_THIS("CRtRudpConn::Release, pdu="<<m_sendPduBuffer.Size()
		<<" buff="<<RUDP_SEND_BUFFER_SIZE
		<<" orderingList="<<orderingList.Size()
		<<" ack="<<acknowlegements.Size()
		<<" resendList="<<resendList.Size()
		<<" resendQueue="<<resendQueue.Size()
		<<" sendPacketSet="<<sendPacketSet[IRtTransport::DP_MEDIUM].Size()
		<<" split="<<splitPacketChannelList.Size()
		<<" hasRecved="<<hasReceivedPacketQueue.Size());
*/
	//////////////////////////////////////////////////////////////////////////

	Close_i();
	ClearSendPduBuffer();

	//////////////////////////////////////////////////////////////////////////

	unsigned i,j;
	RtRudpPacket *internalPacket;

	for (i=0; i < splitPacketChannelList.Size(); i++)
	{
		for (j=0; j < splitPacketChannelList[i]->splitPacketList.Size(); j++)
		{
//			splitPacketChannelList[i]->splitPacketList[j]->data->DestroyChained();
			m_pFlowControl->m_internalPacketPool->Delete( splitPacketChannelList[i]->splitPacketList[j] );
		}
		delete splitPacketChannelList[i];
	}
	splitPacketChannelList.Clear();

	for ( i = 0; i < orderingList.Size(); i++ )
	{
		if ( orderingList[ i ] )
		{
			RudpRecvOrderingListType* theList = orderingList[ i ];
			while ( theList->Size() )
			{
				internalPacket = orderingList[ i ]->Pop();
//				internalPacket->data->DestroyChained();
				m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
			}
			delete theList;
		}
	}
	orderingList.Clear();

	resendList.Clear();
	while ( !resendQueue.IsEmpty() )
	{
		// The resend Queue can have NULL pointer holes.  This is so we can deallocate blocks without having to compress the array
		internalPacket = resendQueue.Pop();

		if ( internalPacket )
		{
//			internalPacket->data->DestroyChained();
			m_pFlowControl->m_internalPacketPool->Delete( internalPacket );
		}
	}
	resendQueue.Clear();

	for ( i = 0; i < DP_NUMBER_OF_PRIORITIES; i++ )
	{
		j = 0;
		for ( ; j < sendPacketSet[ i ].Size(); j++ )
		{
//			( sendPacketSet[ i ] ) [ j ]->data->DestroyChained();
			m_pFlowControl->m_internalPacketPool->Delete( ( sendPacketSet[ i ] ) [ j ] );
		}

		sendPacketSet[ i ].Clear(); // Preallocate the send lists so we don't do a bunch of reallocations unnecessarily
	}

	m_pFlowControl->m_internalPacketPool->ClearPool(128);

	acknowlegements.Clear();
	m_pFlowControl = NULL;
}

void CRtRudpConn::HandleReceivedPacket(RtRudpPacket* aPacket)
{
	switch(aPacket->pduType)
	{
	case RUDP_PDU_CONNREQ:
		HandleConnReqPdu(*aPacket->data);
		break;
	case RUDP_PDU_CONNRESP:
		HandleConnRespPdu(*aPacket->data);
		break;
	case RUDP_PDU_SHUTDOWN:
		HandleShutdownPdu();
		break;
	case RUDP_PDU_KEEPALIVE:
		UpdateSending(lastUpdateTime);
//		RUDP_INFO_TRACE_THIS("CRtRudpConn::HandleReceivedPacket, receive keepalive pdu.");
		break;
	case RUDP_PDU_DATA:
		HandleDataPdu(*aPacket->data);
		break;
	default:
		{
			RUDP_WARNING_TRACE_THIS("CRtRudpConn::HandleReceivedPacket, unknown pdu! type="<<aPacket->pduType<<" state="<<m_status);
			if (m_status==RUDP_STATE_SYNC_RECV || m_status==RUDP_STATE_SYNC_SENT)
			{
				OnDisconnect(RT_ERROR_NETWORK_INVALID_HANDSHAKE, m_lowerTransport.ParaIn());
			}
		}
	}

//	aPacket->data->DestroyChained();
	m_pFlowControl->m_internalPacketPool->Delete(aPacket);
}

void CRtRudpConn::HandleShutdownPdu()
{
/*
	if (m_status!=RUDP_STATE_OPEN)
	{
		RUDP_ERROR_TRACE_THIS("CRtRudpConn::HandleShutdownPdu, unexpected! state="<<m_status);
		return;
	}
*/
	RUDP_INFO_TRACE_THIS("CRtRudpConn::HandleShutdownPdu, state="<<m_status);
	UpdateSending(lastUpdateTime); // Send ack of shutdown pdu.
	OnDisconnect(RT_ERROR_NETWORK_PEER_CLOSE_CONNECTION, m_lowerTransport.ParaIn());
}

void CRtRudpConn::HandleDataPdu(CRtMessageBlock& aData)
{
	if (m_status!=RUDP_STATE_OPEN)
	{
		if (m_status==RUDP_STATE_CONNREQ_SENT)
		{
			CacheReceivedData(aData);
			return;
		}
//		RUDP_ERROR_TRACE_THIS("CRtRudpConn::HandleDataPdu, unexpected! state="<<m_status);
		return;
	}

	if (m_sink)
		m_sink->OnReceive(aData, this);
}

#if 0
void CRtRudpConn::HandleDropPacket(RtRudpPacket* aPacket, RtRudpTimeUS inCurTime)
{
/*
	RUDP_INFO_TRACE_THIS("CRtRudpConn::HandleDropPacket,"
		" packetNo="<<aPacket->packetNo<<
// 		" reliability="<<aPacket->reliability<<
		" baseIndex="<<receivedPacketsBaseIndex<<
		" size="<<hasReceivedPacketQueue.Size());
*/
	++m_recvDropPackets;

	RtRudpPacketNo dropPacketNo = aPacket->packetNo;

	if (aPacket->splitPacketCount>0)
	{
		dropPacketNo = aPacket->packetNo - aPacket->splitPacketIndex + aPacket->splitPacketCount - 1;
		DeleteOldSplitPackets(aPacket->splitPacketId);
	}

// 	if (receivedPacketsBaseIndex<=dropPacketNo)
	if (RUDP_NOT_MORE_THAN(receivedPacketsBaseIndex, dropPacketNo, RtRudpPacketNo))
	{
		if (hasReceivedPacketQueue.Size()>0)
		{
// 			if (receivedPacketsBaseIndex+hasReceivedPacketQueue.Size()-1>dropPacketNo)
			if (!RUDP_NOT_MORE_THAN(receivedPacketsBaseIndex+hasReceivedPacketQueue.Size()-1, dropPacketNo, RtRudpPacketNo))
			{
// 				for (RtRudpPacketNo i=receivedPacketsBaseIndex; i<=dropPacketNo; ++i)
				for (RtRudpPacketNo i=0; i<=dropPacketNo-receivedPacketsBaseIndex; ++i)
				{
					hasReceivedPacketQueue.Pop();
				}
			}
			else {
				hasReceivedPacketQueue.ClearAndForceAllocation(DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE);
			}
		}
		receivedPacketsBaseIndex = dropPacketNo + 1;
	}

	ReceivedPacketQueuePopExpiredTimeAndCompress(inCurTime);

	if (aPacket->reliability==RUDP_RELIABLE_ORDERED)
	{
// 		if (waitingForOrderedPacketReadIndex[aPacket->orderingChannel]<=aPacket->orderingIndex)
		if (RUDP_NOT_MORE_THAN(waitingForOrderedPacketReadIndex[aPacket->orderingChannel], aPacket->orderingIndex, RtRudpOrderingID))
		{
			waitingForOrderedPacketReadIndex[aPacket->orderingChannel] = aPacket->orderingIndex + 1;

			RudpRecvOrderingListType* orderingListAtOrderingStream = GetOrderingListAtOrderingStream( aPacket->orderingChannel );
			if ( orderingListAtOrderingStream != 0)
			{
				while ( orderingListAtOrderingStream->Size() > 0 )
				{
					orderingListAtOrderingStream->Beginning();
					bool indexFound=false;
					int size=orderingListAtOrderingStream->Size();
					int count=0;

					while (count++ < size)
					{
						if ( RUDP_NOT_MORE_THAN(orderingListAtOrderingStream->Peek()->orderingIndex, 
							waitingForOrderedPacketReadIndex[aPacket->orderingChannel], RtRudpOrderingID) )
						{
							RtRudpPacket* internalPacket = orderingListAtOrderingStream->Pop();
							if (internalPacket->orderingIndex == waitingForOrderedPacketReadIndex[aPacket->orderingChannel])
							{
								HandleReceivedPacket(internalPacket);
								waitingForOrderedPacketReadIndex[ aPacket->orderingChannel ]++; 
								indexFound=true;
							}
							else {
								// Drop those packets whose orderingIndex are less than the dropped packet.
//								internalPacket->data->DestroyChained();
								m_pFlowControl->m_internalPacketPool->Delete(internalPacket);
							}
						}
						else
							(*orderingListAtOrderingStream)++;
					}

					if (indexFound==false)
						break;
				}
			}
		}
	}

	m_pFlowControl->m_internalPacketPool->Delete(aPacket);
}
#endif 

void CRtRudpConn::ReceivedPacketQueuePopExpiredTimeAndCompress(RtRudpTimeUS time)
{
	// Pop all expired times.  0 means we got the packet, in which case we don't track this index either.
	while ( hasReceivedPacketQueue.Size()>0 && hasReceivedPacketQueue.Peek() < time )
//	while ( hasReceivedPacketQueue.Size()>0 && hasReceivedPacketQueue.Peek() == 0 )
	{
		if (hasReceivedPacketQueue.Pop()!=0)
			++m_recvDropPackets;

		++receivedPacketsBaseIndex;
	}

	// If the allocated buffer is > DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE and it is 3x greater than the number of elements actually being used
	if (hasReceivedPacketQueue.AllocationSize() > (unsigned int) DEFAULT_HAS_RECEIVED_PACKET_QUEUE_SIZE && hasReceivedPacketQueue.AllocationSize() > hasReceivedPacketQueue.Size() * 3)
		hasReceivedPacketQueue.Compress();
}

void CRtRudpConn::DeleteOldSplitPackets(RtRudpSplitPacketNo splitPacketId)
{
	unsigned i=0,j;

	while (i < splitPacketChannelList.Size())
	{
		if (RUDP_NOT_MORE_THAN(splitPacketChannelList[i]->splitPacketList[0]->splitPacketId, splitPacketId, RtRudpSplitPacketNo))
		{
			for (j=0; j < splitPacketChannelList[i]->splitPacketList.Size(); j++)
			{
//				splitPacketChannelList[i]->splitPacketList[j]->data->DestroyChained();
				m_pFlowControl->m_internalPacketPool->Delete(splitPacketChannelList[i]->splitPacketList[j]);
			}
			delete splitPacketChannelList[i];
			splitPacketChannelList.RemoveAtIndex(i);
		}
		else
			i++;
	}
}

//////////////////////////////////////////////////////////////////////////
// class CRtRudpConnClient
//////////////////////////////////////////////////////////////////////////
CRtRudpConnClient::CRtRudpConnClient(CRtConnectionManager::CType)
: m_sendSyncCount(0)
, m_lastPing(0)
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnClient::CRtRudpConnClient");
	m_needKeepAlive = TRUE;
}

CRtRudpConnClient::~CRtRudpConnClient()
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnClient::~CRtRudpConnClient");
	ClearReceivedData();
}

RtResult CRtRudpConnClient::Disconnect(RtResult inReason)
{
	if (m_status==RUDP_STATE_SYNC_SENT)
		m_sendSyncTimer.Cancel();

	return CRtRudpConn::Disconnect(inReason);
}

void CRtRudpConnClient::OnReceive(
	   CRtMessageBlock &aData,
	   IRtTransport *aTrptId,
	   CRtTransportParameter *aPara)
{
	DWORD dwDataLen = aData.GetChainedLength();

	if (m_status!=RUDP_STATE_SYNC_SENT && dwDataLen>2)
	{
		CRtRudpConn::OnReceive_i(aData, aTrptId, aPara);
		return;
	} 

	if (m_status==RUDP_STATE_SYNC_SENT)
	{
		RtResult rv = RT_ERROR_NETWORK_INVALID_SYNC;
		if (dwDataLen==2)
		{
			LPBYTE lpData = (LPBYTE)aData.GetTopLevelReadPtr();
			if (lpData[0]==(BYTE)RUDP_PDU_SYNC && lpData[1]==RUDP_PACKET_PADDING)
			{
				RUDP_INFO_TRACE_THIS("CRtRudpConnClient::OnReceive, receive sync pdu from server.");
				m_sendSyncTimer.Cancel();

				// Send ConnReq pdu. 
				RakNet::BitStream outBitStream(RUDP_CONNREQ_LEN);
				outBitStream.Write((BYTE)RUDP_VERSION);
				outBitStream.Write((DWORD)RUDP_MAGIC_NUMBER);
				outBitStream.Write(GetLocalIP());
				outBitStream.Write(m_localAddr.GetPort());
				CRtMessageBlock mb(RUDP_CONNREQ_LEN, (LPCSTR)outBitStream.GetData(), CRtMessageBlock::DONT_DELETE, RUDP_CONNREQ_LEN);
				SendPdu(RUDP_PDU_CONNREQ, mb, RUDP_RELIABLE, DP_SYSTEM);
				UpdateSending(GetRtRudpTimeUS());
				SetStatus(RUDP_STATE_CONNREQ_SENT);
#ifdef RUDP_FLOWCONTROL_OPTIMIZE
				InsertIntoFlowControl();
#else
				m_pFlowControl->InsertConn(this);
#endif // RUDP_FLOWCONTROL_OPTIMIZE
				return;
			}

			if (lpData[0]==(BYTE)RUDP_PDU_PUNCH && lpData[1]==RUDP_PACKET_PADDING)
			{
				// Ignore the pdu.
				return;
			}
		}

		RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::OnReceive, invalid sync! len="<<dwDataLen);
		OnDisconnect(rv, aTrptId);
		return;
	}

	if (dwDataLen==2 && aData.GetTopLevelReadPtr()[0]==(char)RUDP_PDU_RESET && aData.GetTopLevelReadPtr()[1]==(char)RUDP_PACKET_PADDING)
	{
		/// It may be the case that client sent data to a different server from what it originally 
		/// connected to, while both servers listened on the same virtual IP and port.
		RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::OnReceive, get RESET pdu! state="<<m_status);
		if (m_status!=RUDP_STATE_CLOSED)
			OnDisconnect(RT_ERROR_NETWORK_SOCKET_RESET, aTrptId);

		return;
	}

	RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::OnReceive, invalid data! state="<<m_status<<" len="<<dwDataLen);
}

void CRtRudpConnClient::OnDisconnect(
		RtResult aReason,
		IRtTransport *aTrptId)
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnClient::OnDisconnect, state="<<m_status<<" reason="<<aReason<<" aTrptId="<<aTrptId);

	if (m_status==RUDP_STATE_CLOSED)
		return;

	// As in m_sink->OnDisconnect(), it may call CRtRudpConn::Disconnect() again, so the 
	// state has to be set to CLOSED in advance. :-(
	WORD lastStatus = m_status;
	SetStatus(RUDP_STATE_CLOSED);

	if (lastStatus==RUDP_STATE_OPEN)
	{
		if (m_sink)
		{
			m_sink->OnDisconnect(aReason, this);
		}
	}
	else {
		RT_ASSERTE(RT_FAILED(aReason));
		m_pConnConnector->GetConnectorSink()->OnConnectIndication(aReason, NULL, m_pConnConnector);
	}

	ClearReceivedData();
	Close_i();
	ReleaseReference();
}
	
void CRtRudpConnClient::OnConnectIndication(
		RtResult aReason,
		IRtTransport *aTrpt,
		IRtAcceptorConnectorId *aRequestId)
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnClient::OnConnectIndication, state="<<m_status<<" reason="<<aReason<<" trpt="<<aTrpt);

	RtResult rv = aReason;
	if (RT_SUCCEEDED(rv))
	{
		m_lowerTransport = aTrpt;
		m_lowerTransport->OpenWithSink(this);
		GetOption(RT_OPT_TRANSPORT_LOCAL_ADDR, &m_localAddr);
		GetOption(RT_OPT_TRANSPORT_PEER_ADDR, &m_peerAddr);
		rv = SendSync();
		if (RT_SUCCEEDED(rv))
		{
			SetStatus(RUDP_STATE_SYNC_SENT);
			CRtTimeValue tv((time_t)RUDP_SEND_SYNC_INTERVAL);
			m_sendSyncTimer.Schedule(this, tv, 1);
			return;
		}

		RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::OnConnectIndication, send sync pdu failed! rv="<<rv);
	}

	OnDisconnect(rv, m_lowerTransport.ParaIn());
}

void CRtRudpConnClient::OnTimer(CRtTimerWrapper* aId)
{
	if (aId==&m_sendSyncTimer)
	{
//		RUDP_INFO_TRACE_THIS("CRtRudpConnClient::OnTimer, send sync again");
		
		RT_ASSERTE(m_status==RUDP_STATE_SYNC_SENT);
		SendSync();
		CRtTimeValue tv((time_t)RUDP_SEND_SYNC_INTERVAL);
// 		if (m_sendSyncCount%6 == 0)
// 			tv.Set(6.0);
		m_sendSyncTimer.Schedule(this, tv, 1);
		return;
	}

	if (m_status==RUDP_STATE_OPEN && ping>=50 &&
		( ping>1000 || (ping>m_lastPing && ping-m_lastPing>m_lastPing/5) || (ping<m_lastPing && m_lastPing-ping>m_lastPing/5) ) )
	{
		RUDP_INFO_TRACE_THIS("CRtRudpConnClient::OnTimer, rtt="<<ping<<" BuffSize="<<RUDP_SEND_BUFFER_SIZE);
		m_lastPing = ping;
	}

	CRtRudpConn::OnTimer(aId);
}

void CRtRudpConnClient::CancelHandShake()
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnClient::CancelHandShake, state="<<m_status);
	if (m_status!=RUDP_STATE_OPEN && m_status!=RUDP_STATE_CLOSED)
	{
		Disconnect(RT_OK);
	}
}

void CRtRudpConnClient::ScheduleConnectTimer(const CRtTimeValue& aTimeout)
{
	RT_ASSERTE(m_status==RUDP_STATE_INIT);
	m_timer.Schedule(this, aTimeout, 1);
}

RtResult CRtRudpConnClient::SendSync()
{
	BYTE sync[2] = {0};
	sync[0] = (BYTE)RUDP_PDU_SYNC;
	sync[1] = (BYTE)RUDP_PACKET_PADDING;
	++m_sendSyncCount;
	return SendData_i(2, sync);
}

void CRtRudpConnClient::Close_i()
{
	m_sendSyncTimer.Cancel();
	CRtRudpConn::Close_i();
}

void CRtRudpConnClient::HandleConnRespPdu(CRtMessageBlock& aData)
{
	if (m_status==RUDP_STATE_CONNREQ_SENT)
	{
		if (aData.GetChainedLength()==2)
		{
			LPCSTR lpConnResp = aData.GetTopLevelReadPtr();
			if (lpConnResp[0]==(char)RUDP_VERSION)
			{
				if (lpConnResp[1]==(char)RT_OK)
				{
					RUDP_INFO_TRACE_THIS("CRtRudpConnClient::HandleConnRespPdu, handshake successfully.");

					CRtTimeValue tv((time_t)RUDP_KEEPALIVE_INTERVAL);
					m_timer.Schedule(this, tv);
					SetStatus(RUDP_STATE_OPEN);
					m_pConnConnector->GetConnectorSink()->OnConnectIndication(RT_OK, this, m_pConnConnector);
					if (!m_receivedData.empty())
						ClearReceivedData(m_sink);
				}
				else {
					OnDisconnect(RT_ERROR_NETWORK_PEER_REJECT_CONNECTION, m_lowerTransport.ParaIn());
				}
				return;
			}

			RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::HandleConnRespPdu, handshake failed! version="<<lpConnResp[0]);
		}
// 		else
// 			RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::HandleConnReqPdu, invalid length! length="<<aData.GetChainedLength());

		OnDisconnect(RT_ERROR_NETWORK_INVALID_HANDSHAKE, m_lowerTransport.ParaIn());
		return;
	}

	RUDP_WARNING_TRACE_THIS("CRtRudpConnClient::HandleConnRespPdu, state="<<m_status<<" length="<<aData.GetChainedLength());
}

void CRtRudpConnClient::CacheReceivedData(CRtMessageBlock& aData)
{
	m_receivedData.push_back(aData.DuplicateChained());
}

void CRtRudpConnClient::ClearReceivedData(IRtTransportSink* aTransportSink)
{
	for (list<CRtMessageBlock*>::iterator it=m_receivedData.begin(); it!=m_receivedData.end(); ++it)
	{
		if (aTransportSink)
			aTransportSink->OnReceive(*(*it), this);
		(*it)->DestroyChained();
	}
	m_receivedData.clear();
}

//////////////////////////////////////////////////////////////////////////
// class CRtRudpConnServer
//////////////////////////////////////////////////////////////////////////
static int s_rudpConnNum = 0;

CRtRudpConnServer::CRtRudpConnServer(CRtConnectionManager::CType)
{
// 	RUDP_INFO_TRACE_THIS("CRtRudpConnServer::CRtRudpConnServer");

	CRtTimeValue tv((time_t)30);
	m_timer.Schedule(this, tv, 1);
	++s_rudpConnNum;
}

CRtRudpConnServer::~CRtRudpConnServer()
{
	--s_rudpConnNum;
	RUDP_INFO_TRACE_THIS("~CRtRudpConnServer, num="<<s_rudpConnNum);
}

void CRtRudpConnServer::OnReceive(
	   CRtMessageBlock &aData,
	   IRtTransport *aTrptId,
	   CRtTransportParameter *aPara)
{
	DWORD dwDataLen = aData.GetChainedLength();

	if (m_status!=RUDP_STATE_INIT && dwDataLen>2)
	{
		CRtRudpConn::OnReceive_i(aData, aTrptId, aPara);
		return;
	} 

	if (m_status==RUDP_STATE_INIT)
	{
		RtResult rv = RT_ERROR_NETWORK_INVALID_SYNC;
		BOOL bSendReset = TRUE;
		if (dwDataLen==2)
		{
			LPBYTE lpData = (LPBYTE)aData.GetTopLevelReadPtr();
			if (lpData[0]==(BYTE)RUDP_PDU_SYNC && lpData[1]==RUDP_PACKET_PADDING)
			{
				// Response with SYNC pdu too. 
				rv = SendData_i(2, lpData);
				if (RT_SUCCEEDED(rv))
				{
					SetStatus(RUDP_STATE_SYNC_RECV);
					return;
				}

				RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::OnReceive, send sync pdu failed! rv="<<rv);
			}
			else if (lpData[0]==(BYTE)RUDP_PDU_RESET && lpData[1]==RUDP_PACKET_PADDING)
			{
				// It could happen in P2P connection case. 
				RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::OnReceive, get RESET pdu! state="<<m_status);
				bSendReset = FALSE;
			}
		}

		if (bSendReset)
		{
// 			RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::OnReceive, reset connection! state="<<m_status<<" len="<<dwDataLen);
			BYTE resetPdu[2] = {0};
			resetPdu[0] = RUDP_PDU_RESET;
			resetPdu[1] = RUDP_PACKET_PADDING;
			SendData_i(2, resetPdu);
		}

		OnDisconnect(rv, aTrptId);
		return;
	}

	if (dwDataLen==2 && aData.GetTopLevelReadPtr()[0]==(char)RUDP_PDU_RESET && aData.GetTopLevelReadPtr()[1]==(char)RUDP_PACKET_PADDING)
	{
		/// It could happen in P2P connection case. 
		RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::OnReceive, get RESET pdu! state="<<m_status);
		if (m_status!=RUDP_STATE_CLOSED)
			OnDisconnect(RT_ERROR_NETWORK_SOCKET_RESET, aTrptId);

		return;
	}

// 	RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::OnReceive, invalid data! state="<<m_status<<" len="<<dwDataLen);
}

void CRtRudpConnServer::OnDisconnect(
		RtResult aReason,
		IRtTransport *aTrptId)
{
	RUDP_INFO_TRACE_THIS("CRtRudpConnServer::OnDisconnect, state="<<m_status<<" reason="<<aReason<<" aTrptId="<<aTrptId);
	
	if (m_status==RUDP_STATE_CLOSED)
		return;

	// As in m_sink->OnDisconnect(), it may call CRtRudpConn::Disconnect() again, so the 
	// state has to be set to CLOSED in advance. :-(
	WORD lastStatus = m_status;
	SetStatus(RUDP_STATE_CLOSED);
	
	if (lastStatus==RUDP_STATE_OPEN && m_sink)
	{
		m_sink->OnDisconnect(aReason, this);
	}
	Close_i();
	ReleaseReference();
}
	
void CRtRudpConnServer::OnConnectIndication(
		RtResult aReason,
		IRtTransport *aTrpt,
		IRtAcceptorConnectorId *aRequestId)
{
	RT_ASSERTE(m_status == RUDP_STATE_INIT);
	RT_ASSERTE(RT_SUCCEEDED(aReason));
	RT_ASSERTE(m_pConnAcceptor->GetTPAcceptor() == aRequestId);

	m_lowerTransport = aTrpt;
	aTrpt->OpenWithSink(this);
	GetOption(RT_OPT_TRANSPORT_LOCAL_ADDR, &m_localAddr);
	GetOption(RT_OPT_TRANSPORT_PEER_ADDR, &m_peerAddr);
// 	RUDP_INFO_TRACE_THIS("CRtRudpConnServer::OnConnectIndication, IP="<<m_peerAddr.GetIpDisplayName()<<" port="<<m_peerAddr.GetPort()<<" num="<<s_rudpConnNum);
}

void CRtRudpConnServer::HandleConnReqPdu(CRtMessageBlock& aData)
{
	if (m_status==RUDP_STATE_SYNC_RECV)
	{
		BYTE connResp[2] = {0};
		connResp[0] = RUDP_VERSION;
		if (aData.GetChainedLength()==RUDP_CONNREQ_LEN)
		{
			RakNet::BitStream inBitStream( (unsigned char *) aData.GetTopLevelReadPtr(), aData.GetTopLevelLength(), false );
			BYTE version = 0;
			DWORD magicNumber = 0;

			inBitStream.Read(version);
			if (version==RUDP_VERSION)
			{
				inBitStream.Read(magicNumber);
				if (magicNumber==(DWORD)RUDP_MAGIC_NUMBER)
				{
					RUDP_INFO_TRACE_THIS("CRtRudpConnServer::HandleConnReqPdu, PeerIP="
						<<m_peerAddr.GetIpDisplayName()
						<<" Port="<<m_peerAddr.GetPort()
						<<" LocalIP="<<m_localAddr.GetIpDisplayName()
						<<" Port="<<m_localAddr.GetPort()
						<<" num="<<s_rudpConnNum);

					DWORD peerPrivateIP = 0;
					WORD peerPrivatePort = 0;
					inBitStream.Read(peerPrivateIP);
					inBitStream.Read(peerPrivatePort);
					m_peerPrivateAddr.SetIpAddrBy4Bytes(peerPrivateIP);
					m_peerPrivateAddr.SetPort(peerPrivatePort);

					connResp[1] = RT_OK;
					CRtMessageBlock mb(2, (LPCSTR)connResp, CRtMessageBlock::DONT_DELETE, 2);
					SendPdu(RUDP_PDU_CONNRESP, mb, RUDP_RELIABLE, DP_SYSTEM);
					UpdateSending(GetRtRudpTimeUS());
#ifdef RUDP_FLOWCONTROL_OPTIMIZE
					InsertIntoFlowControl();
#else
					m_pFlowControl->InsertConn(this);
#endif // RUDP_FLOWCONTROL_OPTIMIZE

					CRtTimeValue tv((time_t)RUDP_KEEPALIVE_INTERVAL);
					m_timer.Schedule(this, tv);
					SetStatus(RUDP_STATE_OPEN);
					m_pConnAcceptor->GetAcceptorSink()->OnConnectIndication(RT_OK, this, m_pConnAcceptor.ParaIn());
					return;
				}
			}
			RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::HandleConnReqPdu, handshake failed!");
		}
// 		else
// 			RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::HandleConnReqPdu, invalid length! length="<<aData.GetChainedLength());

		connResp[1] = (BYTE)RT_ERROR_NETWORK_INVALID_HANDSHAKE;
		CRtMessageBlock mb(2, (LPCSTR)connResp, CRtMessageBlock::DONT_DELETE, 2);
		SendPdu(RUDP_PDU_CONNRESP, mb, RUDP_UNRELIABLE, DP_SYSTEM);
		UpdateSending(GetRtRudpTimeUS());
		OnDisconnect(RT_ERROR_NETWORK_INVALID_HANDSHAKE, m_lowerTransport.ParaIn());
		return;
	}

	RUDP_WARNING_TRACE_THIS("CRtRudpConnServer::HandleConnReqPdu, state="<<m_status<<" length="<<aData.GetChainedLength());
}

